package com.chis.netty.mqtt.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttIdentifierRejectedException;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageFactory;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnacceptableProtocolVersionException;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.StringUtils;

import java.util.HashMap;
import java.util.Map;


public class MqttServerHandler extends SimpleChannelInboundHandler<MqttMessage>{
    private MqttVersion version;
    private String clientId;
    private String userName;
    private String brokerId;
    private boolean connected;
    private boolean cleanSession;
    private MqttPublishMessage willMessage;

    public MqttServerHandler() {
    }


    @Override
    //channelHandlerContext 通首的上下文  mqttMessage 客户端来的MQTT报文
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, MqttMessage mqttMessage) throws Exception {
        if (mqttMessage.decoderResult().isFailure()) {
            //mqtt消息接收失败
            Throwable cause = mqttMessage.decoderResult().cause();
            if (cause instanceof MqttUnacceptableProtocolVersionException) {
                BrokerSessionHelper.sendMessage(
                        channelHandlerContext,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false),
                                null),
                        "INVALID",
                        null,
                        true);

            } else if (cause instanceof MqttIdentifierRejectedException) {
                BrokerSessionHelper.sendMessage(
                        channelHandlerContext,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                                null),
                        "INVALID",
                        null,
                        true);
            }
            channelHandlerContext.close();

            return;
        }

        switch (mqttMessage.fixedHeader().messageType()) {
            case CONNECT:
                // 客户端连接连接服务端后的第一个报文，
                onConnect(channelHandlerContext, (MqttConnectMessage) mqttMessage);
                break;
            case PUBLISH:
                // 发布消息
                onPublish(channelHandlerContext, (MqttPublishMessage) mqttMessage);
                break;
            case PUBACK:
                // 发布确认
                onPubAck(channelHandlerContext, mqttMessage);
                break;
            case PUBREC:
                // 发布收到
                onPubRec(channelHandlerContext, mqttMessage);
                break;
            case PUBREL:
                // 发布释放
                onPubRel(channelHandlerContext, mqttMessage);
                break;
            case PUBCOMP:
                // 发布完成
                onPubComp(channelHandlerContext, mqttMessage);
                break;
            case SUBSCRIBE:
                 // 订阅主题
                onSubscribe(channelHandlerContext, (MqttSubscribeMessage) mqttMessage);
                break;
            case UNSUBSCRIBE:
                // 取消订阅
                onUnsubscribe(channelHandlerContext, (MqttUnsubscribeMessage) mqttMessage);
                break;
            case PINGREQ:
                // 心跳请求
                onPingReq(channelHandlerContext);
                break;
            case DISCONNECT:
                // 断开连接
                onDisconnect(channelHandlerContext);
                break;
        }

    }

    private void onConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
        System.out.println("CONNECT:"+ msg.payload());

        this.version = MqttVersion.fromProtocolNameAndLevel(msg.variableHeader().name(), (byte) msg.variableHeader().version());
        this.clientId = msg.payload().clientIdentifier();
        this.cleanSession = msg.variableHeader().isCleanSession();

        //MQTT 3.1之后可能存在为空的客户ID。所以要进行处理。如果客户ID是空，而且还在保存处理相关的信息。这样子是不行。
        //必须有客户ID我们才能存保相关信息。
        if (StringUtils.isBlank(this.clientId)) {
            if (!this.cleanSession) {
                BrokerSessionHelper.sendMessage(
                        ctx,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                                null),
                        "INVALID",
                        null,
                        true);
                ctx.close();
                return;
            } else {
                this.clientId =  java.util.UUID.randomUUID().toString();
            }
        }

        //有可能发送俩次的连接包。如果已经存在连接就是关闭当前的连接。
        if (this.connected) {
            ctx.close();
            return;
        }

        boolean userNameFlag = msg.variableHeader().hasUserName();
        boolean passwordFlag = msg.variableHeader().hasPassword();
        this.userName = msg.payload().userName();

        String password = "" ;
        if( msg.payload().passwordInBytes() != null  && msg.payload().passwordInBytes().length > 0)
            password =   new String(msg.payload().passwordInBytes());

        boolean mistake = false;

        //如果有用户名标示，那么就必须有密码标示。
        //当有用户名标的时候，用户不能为空。
        if (userNameFlag) {
            if (StringUtils.isBlank(this.userName))
                mistake = true;
        } else {
            if (StringUtils.isNotBlank(this.userName) || passwordFlag) mistake = true;
        }

        //当有密码标示的时候，密码不能为空。
        if (passwordFlag) {
            if (StringUtils.isBlank(password)) mistake = true;
        } else {
            if (StringUtils.isNotBlank(password)) mistake = true;
        }

        if (mistake) {
            BrokerSessionHelper.sendMessage(
                    ctx,
                    MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                            new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false),
                            null),
                    this.clientId,
                    null,
                    true);
            ctx.close();
            return;
        }

        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, !this.cleanSession),
                        null),
                this.clientId,
                null,
                true);

        ChannelHandlerContext lastSession = BrokerSessionHelper.removeSession(this.clientId);
        if (lastSession != null) {
            lastSession.close();
        }

        String willTopic = msg.payload().willTopic();
        String willMessage = "";
        if(msg.payload().willMessageInBytes() != null && msg.payload().willMessageInBytes().length > 0)
            willMessage =  new String(msg.payload().willMessageInBytes());

        if (msg.variableHeader().isWillFlag() && StringUtils.isNotEmpty(willTopic) && StringUtils.isNotEmpty(willMessage)) {

            this.willMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(msg.variableHeader().willQos()), msg.variableHeader().isWillRetain(), 0),
                    new MqttPublishVariableHeader(willTopic, 0),
                    Unpooled.wrappedBuffer(willMessage.getBytes())
            );
        }

        this.connected = true;
        BrokerSessionHelper.saveSession(this.clientId, ctx);
    }

    private void onSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
        System.out.println("SUB");
    }

    private void onUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) {
        System.out.println("UNSUB");
    }

    private void onPingReq(ChannelHandlerContext ctx) {
        System.out.println("PINGREQ");
    }

    private void onDisconnect(ChannelHandlerContext ctx) {
        System.out.println("DISCONNECT");
        if (!this.connected) {
            ctx.close();
            return;
        }
        BrokerSessionHelper.removeSession(this.clientId, ctx);
        this.willMessage = null;
        this.connected = false;
        ctx.close();

    }

    private void onPubComp(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println("PUBCOMP");

    }

    private void onPubRel(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println("PUBREL");
    }

    private void onPubRec(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println("PUBREC");
    }

    private void onPubAck(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println("PUBACK:");
    }

    private void onPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
        System.out.println("PUBLISH");

        ByteBuf buf1 = Unpooled.wrappedBuffer((ByteBuf)msg.payload());
        System.out.println("接收客户端数据: "+buf1.toString(CharsetUtil.UTF_8));

        //调用服务
        String res = invokeService();
        //向客户端发送结果
        byte[] bytes = res.getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = Unpooled.wrappedBuffer(bytes);

        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttPublishVariableHeader("topic_aa", 1),
                        buf),
                "INVALID",
                null,
                true);
    }

	@Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
	    //服务端超时的逻辑处理
        System.out.println("已经5秒未收到客户端的消息了！");
        if(evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent) evt;
            System.out.println("超时信息："+ctx.channel().remoteAddress() + " , " +event.state());
            if(event.state() == IdleState.READER_IDLE){
                ChannelFuture future = ctx.channel().writeAndFlush("客户端超时关闭连接！");
                future.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        //消息写入成功，关闭通道
                        future.channel().close();
                    }
                });
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    /**
     * 调用服务
     * @return
     */
    private String invokeService() {
        return "服务调用成功";
    }

}
